package com.hrt.bigdata.mr.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 执行该代码不需要四个配置文件
 */
public class DriverOnLocal {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Path inputPath=new Path("F:\\P8\\workspace\\hadoop-p8\\data\\*");
        Path outputPath=new Path("F:\\P8\\workspace\\hadoop-p8\\data\\result");

        //作为整个Job的配置
        Configuration conf = new Configuration();
        //保证输出目录不存在
        FileSystem fs=FileSystem.get(conf);
        if (fs.exists(outputPath)) {
            fs.delete(outputPath, true);
        }
        // ①创建Job
        Job job = Job.getInstance(conf);
        // 将某个类所在地jar包作为job的jar包
        job.setJarByClass(DriverOnLocal.class);

        // 为Job创建一个名字
        job.setJobName("wordcount");
        // ②设置Job
        // 设置Job运行的Mapper，Reducer类型，Mapper,Reducer输出的key-value类型
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);
        // Job需要根据Mapper和Reducer输出的Key-value类型准备序列化器，通过序列化器对输出的key-value进行序列化和反序列化
        // 如果Mapper和Reducer输出的Key-value类型一致，直接设置Job最终的输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 设置输入目录和输出目录
        FileInputFormat.setInputPaths(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);
        // ③运行Job
        job.waitForCompletion(true);
    }
}
